package com.niit.adverStat

/**
 * @author Liweijian.
 * @Description jdbc连接工具类
 * @date 2025/07/11 14:49
 */

import com.niit.commons.pool.{CreateMySqlPool, MySqlProxy, QueryCallback} // 确保 MySqlProxy 被导入

import java.sql.ResultSet
import scala.collection.mutable.ArrayBuffer

// *** 【重要】AdBlacklist case class 定义 - 确保与 AdverStat.scala 中的定义一致 ***
// 如果此 case class 在另一个文件中（例如 commons.constant），请确保在这里导入
// case class AdBlacklist(userid: Long, blacklistLevel: Int) // 注意：blacklistLevel 为 Int 类型

/**
 * 用户黑名单DAO类
 */
object AdBlacklistDAO {

  /**
   * 批量插入/更新广告黑名单用户，支持等级提升
   *
   * @param adBlacklists 待插入/更新的黑名单用户列表
   */
  def insertBatch(adBlacklists: Array[AdBlacklist]): Unit = {
    var client: MySqlProxy = null // 获取 MySqlProxy 实例

    try {
      val mySqlPool = CreateMySqlPool() // 获取对象池单例对象
      client = mySqlPool.borrowObject() // 从对象池中提取对象

      // SQL 语句：如果 userid 已存在，则更新 blacklist_level
      // 使用 GREATEST 函数，因为等级是数字，直接取最大值即可实现等级提升
      val sql =
        """
          INSERT INTO ad_blacklist (userid, blacklist_level)
          VALUES (?, ?)
          ON DUPLICATE KEY UPDATE
          blacklist_level = GREATEST(blacklist_level, VALUES(blacklist_level));
        """.stripMargin

      // 构建批量参数
      val paramsList = new ArrayBuffer[Array[Any]]()
      adBlacklists.foreach { adBlacklist =>
        paramsList += Array[Any](adBlacklist.userid, adBlacklist.blacklistLevel) // 传递 userid 和 blacklistLevel
      }

      // 执行批量插入/更新操作
      val results = client.executeBatch(sql, paramsList.toArray)
      println(s"AdBlacklistDAO: 成功批量插入/更新 ${adBlacklists.size} 条黑名单记录。批处理结果: ${results.mkString(",")}")

    } catch {
      case e: Exception =>
        println(s"AdBlacklistDAO: 批量插入/更新 ad_blacklist 时发生异常: ${e.getMessage}")
        e.printStackTrace()
    } finally {
      // 确保将 MySqlProxy 实例返回给对象池，无论是否发生异常
      if (client != null) {
        CreateMySqlPool().returnObject(client)
      }
    }
  }

  /**
   * 查询所有广告黑名单用户及其等级
   *
   * @return 包含用户ID和黑名单等级的 AdBlacklist 数组
   */
  def findAll(): Array[AdBlacklist] = {
    val adBlacklists = new ArrayBuffer[AdBlacklist]()
    var client: MySqlProxy = null // 获取 MySqlProxy 实例

    try {
      val mySqlPool = CreateMySqlPool() // 获取对象池单例对象
      client = mySqlPool.borrowObject() // 从对象池中提取对象

      val sql = "SELECT userid, blacklist_level FROM ad_blacklist" // 查询 blacklist_level 列

      // 执行 SQL 查询并处理结果集
      client.executeQuery(sql, null, new QueryCallback {
        override def process(rs: ResultSet): Unit = {
          while (rs.next()) {
            val userid = rs.getLong("userid") // 按列名获取，更健壮
            val blacklistLevel = rs.getInt("blacklist_level") // 获取 Int 类型的黑名单等级
            adBlacklists += AdBlacklist(userid, blacklistLevel)
          }
        }
      })
      println(s"AdBlacklistDAO: 成功查询 ${adBlacklists.size} 条黑名单记录。")

    } catch {
      case e: Exception =>
        println(s"AdBlacklistDAO: 查询 ad_blacklist 时发生异常: ${e.getMessage}")
        e.printStackTrace()
    } finally {
      // 确保将 MySqlProxy 实例返回给对象池
      if (client != null) {
        CreateMySqlPool().returnObject(client)
      }
    }
    adBlacklists.toArray
  }
}


/**
 * 用户广告点击量DAO实现类
 *
 */
object AdUserClickCountDAO {

  def updateBatch(adUserClickCounts: Array[AdUserClickCount]): Unit = {
    var client: MySqlProxy = null // 获取 MySqlProxy 实例
    try {
      val mySqlPool = CreateMySqlPool()
      client = mySqlPool.borrowObject()

      // 首先对用户广告点击量进行分类，分成待插入的和待更新的
      val insertAdUserClickCounts = ArrayBuffer[AdUserClickCount]()
      val updateAdUserClickCounts = ArrayBuffer[AdUserClickCount]()

      val selectSQL = "SELECT count(*) FROM ad_user_click_count WHERE date=? AND userid=? AND adid=? "

      // 这里需要在单个分区内迭代，而不是在RDD层面，因为client是每个分区获取一次的
      // 如果 adUserClickCounts 很大，这种每个元素都查一次数据库的方式效率不高
      // 更好的方式是先将所有 adUserClickCounts 的键 (date, userid, adid) 收集起来
      // 进行批量查询，然后根据查询结果分类。但为了保持原逻辑，这里先不动。
      for (adUserClickCount <- adUserClickCounts) {
        val selectParams: Array[Any] = Array(adUserClickCount.date, adUserClickCount.userid, adUserClickCount.adid)
        // 根据传入的用户点击次数统计数据从已有的ad_user_click_count中进行查询
        client.executeQuery(selectSQL, selectParams, new QueryCallback {
          override def process(rs: ResultSet): Unit = {
            // 如果能查询到并且点击次数大于0，则认为是待更新项
            if (rs.next() && rs.getInt(1) > 0) {
              updateAdUserClickCounts += adUserClickCount
            } else {
              insertAdUserClickCounts += adUserClickCount
            }
          }
        })
      }

      // 执行批量插入
      val insertSQL = "INSERT INTO ad_user_click_count VALUES(?,?,?,?)"
      val insertParamsList: ArrayBuffer[Array[Any]] = ArrayBuffer[Array[Any]]()

      // 将待插入项全部加入到参数列表中
      for (adUserClickCount <- insertAdUserClickCounts) {
        insertParamsList += Array[Any](adUserClickCount.date, adUserClickCount.userid, adUserClickCount.adid, adUserClickCount.clickCount)
      }
      if (insertParamsList.nonEmpty) {
        client.executeBatch(insertSQL, insertParamsList.toArray)
        println(s"AdUserClickCountDAO: 成功批量插入 ${insertParamsList.size} 条记录。")
      }


      // 执行批量更新
      // clickCount=clickCount + ：此处的UPDATE是进行累加
      val updateSQL = "UPDATE ad_user_click_count SET clickCount=clickCount + ? WHERE date=? AND userid=? AND adid=?"
      val updateParamsList: ArrayBuffer[Array[Any]] = ArrayBuffer[Array[Any]]()

      // 将待更新项全部加入到参数列表中
      for (adUserClickCount <- updateAdUserClickCounts) {
        updateParamsList += Array[Any](adUserClickCount.clickCount, adUserClickCount.date, adUserClickCount.userid, adUserClickCount.adid)
      }
      if (updateParamsList.nonEmpty) {
        client.executeBatch(updateSQL, updateParamsList.toArray)
        println(s"AdUserClickCountDAO: 成功批量更新 ${updateParamsList.size} 条记录。")
      }

    } catch {
      case e: Exception =>
        println(s"AdUserClickCountDAO: 批量更新/插入时发生异常: ${e.getMessage}")
        e.printStackTrace()
    } finally {
      // 确保将 MySqlProxy 实例返回给对象池
      if (client != null) {
        CreateMySqlPool().returnObject(client)
      }
    }
  }

  /**
   * 根据多个key查询用户广告点击量
   *
   * @param date   日期
   * @param userid 用户id
   * @param adid   广告id
   * @return
   */
  def findClickCountByMultiKey(date: String, userid: Long, adid: Long): Int = {
    var client: MySqlProxy = null // 获取 MySqlProxy 实例
    var clickCount = 0

    try {
      val mySqlPool = CreateMySqlPool()
      client = mySqlPool.borrowObject()

      val sql = "SELECT clickCount FROM ad_user_click_count " +
        "WHERE date=? " +
        "AND userid=? " +
        "AND adid=?"

      val params = Array[Any](date, userid, adid)

      // 根据多个条件查询指定用户的点击量，将查询结果累加到clickCount中
      client.executeQuery(sql, params, new QueryCallback {
        override def process(rs: ResultSet): Unit = {
          if (rs.next()) {
            clickCount = rs.getInt(1)
          }
        }
      })
      // println(s"AdUserClickCountDAO: 查询用户 ${userid} 广告 ${adid} 在 ${date} 的点击量为: ${clickCount}") // 可能会很频繁，调试时开启

    } catch {
      case e: Exception =>
        println(s"AdUserClickCountDAO: 查询 findClickCountByMultiKey 时发生异常: ${e.getMessage}")
        e.printStackTrace()
    } finally {
      // 确保将 MySqlProxy 实例返回给对象池
      if (client != null) {
        CreateMySqlPool().returnObject(client)
      }
    }
    clickCount
  }
}


/**
 * 广告实时统计DAO实现类
 *
 * @author Administrator
 *
 */
object AdStatDAO {

  def updateBatch(adStats: Array[AdStat]): Unit = {
    var client: MySqlProxy = null // 获取 MySqlProxy 实例
    try {
      val mySqlPool = CreateMySqlPool()
      client = mySqlPool.borrowObject()


      // 区分开来哪些是要插入的，哪些是要更新的
      val insertAdStats = ArrayBuffer[AdStat]()
      val updateAdStats = ArrayBuffer[AdStat]()

      val selectSQL = "SELECT count(*) " +
        "FROM ad_stat " +
        "WHERE date=? " +
        "AND province=? " +
        "AND city=? " +
        "AND adid=?"

      for (adStat <- adStats) {
        val params = Array[Any](adStat.date, adStat.province, adStat.city, adStat.adid)
        // 通过查询结果判断当前项时待插入还是待更新
        client.executeQuery(selectSQL, params, new QueryCallback {
          override def process(rs: ResultSet): Unit = {
            if (rs.next() && rs.getInt(1) > 0) {
              updateAdStats += adStat
            } else {
              insertAdStats += adStat
            }
          }
        })
      }

      // 对于需要插入的数据，执行批量插入操作
      val insertSQL = "INSERT INTO ad_stat VALUES(?,?,?,?,?)"

      val insertParamsList: ArrayBuffer[Array[Any]] = ArrayBuffer[Array[Any]]()

      for (adStat <- insertAdStats) {
        insertParamsList += Array[Any](adStat.date, adStat.province, adStat.city, adStat.adid, adStat.clickCount)
      }
      if (insertParamsList.nonEmpty) {
        client.executeBatch(insertSQL, insertParamsList.toArray)
        println(s"AdStatDAO: 成功批量插入 ${insertParamsList.size} 条记录。")
      }


      // 对于需要更新的数据，执行批量更新操作
      // 此处的UPDATE是进行覆盖
      val updateSQL = "UPDATE ad_stat SET clickCount=? " +
        "WHERE date=? " +
        "AND province=? " +
        "AND city=? " +
        "AND adid=?"

      val updateParamsList: ArrayBuffer[Array[Any]] = ArrayBuffer[Array[Any]]()

      for (adStat <- updateAdStats) {
        updateParamsList += Array[Any](adStat.clickCount, adStat.date, adStat.province, adStat.city, adStat.adid)
      }
      if (updateParamsList.nonEmpty) {
        client.executeBatch(updateSQL, updateParamsList.toArray)
        println(s"AdStatDAO: 成功批量更新 ${updateParamsList.size} 条记录。")
      }

    } catch {
      case e: Exception =>
        println(s"AdStatDAO: 批量更新/插入时发生异常: ${e.getMessage}")
        e.printStackTrace()
    } finally {
      // 确保将 MySqlProxy 实例返回给对象池
      if (client != null) {
        CreateMySqlPool().returnObject(client)
      }
    }
  }

}


/**
 * 各省份top3热门广告DAO实现类
 *
 * @author Administrator
 *
 */
object AdProvinceTop3DAO {

  def updateBatch(adProvinceTop3s: Array[AdProvinceTop3]) {
    var client: MySqlProxy = null // 获取 MySqlProxy 实例
    try {
      val mySqlPool = CreateMySqlPool()
      client = mySqlPool.borrowObject()

      // dateProvinces可以实现一次去重
      // AdProvinceTop3：date province adid clickCount，由于每条数据由date province adid组成
      // 当只取date province时，一定会有重复的情况
      val dateProvinces = ArrayBuffer[String]()

      for (adProvinceTop3 <- adProvinceTop3s) {
        // 组合新key
        val key = adProvinceTop3.date + "_" + adProvinceTop3.province

        // dateProvinces中不包含当前key才添加
        // 借此去重
        if (!dateProvinces.contains(key)) {
          dateProvinces += key
        }
      }

      // 根据去重后的date和province，进行批量删除操作
      // 先将原来的数据全部删除
      val deleteSQL = "DELETE FROM ad_province_top3 WHERE date=? AND province=?"

      val deleteParamsList: ArrayBuffer[Array[Any]] = ArrayBuffer[Array[Any]]()

      for (dateProvince <- dateProvinces) {
        val dateProvinceSplited = dateProvince.split("_")
        val date = dateProvinceSplited(0)
        val province = dateProvinceSplited(1)

        val params = Array[Any](date, province)
        deleteParamsList += params
      }
      if (deleteParamsList.nonEmpty) {
        client.executeBatch(deleteSQL, deleteParamsList.toArray)
        println(s"AdProvinceTop3DAO: 成功批量删除 ${deleteParamsList.size} 条记录。")
      }


      // 批量插入传入进来的所有数据
      val insertSQL = "INSERT INTO ad_province_top3 VALUES(?,?,?,?)"

      val insertParamsList: ArrayBuffer[Array[Any]] = ArrayBuffer[Array[Any]]()

      // 将传入的数据转化为参数列表
      for (adProvinceTop3 <- adProvinceTop3s) {
        insertParamsList += Array[Any](adProvinceTop3.date, adProvinceTop3.province, adProvinceTop3.adid, adProvinceTop3.clickCount)
      }
      if (insertParamsList.nonEmpty) {
        client.executeBatch(insertSQL, insertParamsList.toArray)
        println(s"AdProvinceTop3DAO: 成功批量插入 ${insertParamsList.size} 条记录。")
      }

    } catch {
      case e: Exception =>
        println(s"AdProvinceTop3DAO: 批量更新时发生异常: ${e.getMessage}")
        e.printStackTrace()
    } finally {
      // 确保将 MySqlProxy 实例返回给对象池
      if (client != null) {
        CreateMySqlPool().returnObject(client)
      }
    }
  }

}


/**
 * 广告点击趋势DAO实现类
 *
 * @author Administrator
 *
 */
object AdClickTrendDAO {

  def updateBatch(adClickTrends: Array[AdClickTrend]) {
    var client: MySqlProxy = null // 获取 MySqlProxy 实例
    try {
      val mySqlPool = CreateMySqlPool()
      client = mySqlPool.borrowObject()

      // 区分开来哪些是要插入的，哪些是要更新的
      val updateAdClickTrends = ArrayBuffer[AdClickTrend]()
      val insertAdClickTrends = ArrayBuffer[AdClickTrend]()

      val selectSQL = "SELECT count(*) " +
        "FROM ad_click_trend " +
        "WHERE date=? " +
        "AND hour=? " +
        "AND minute=? " +
        "AND adid=?"

      for (adClickTrend <- adClickTrends) {
        // 通过查询结果判断当前项时待插入还是待更新
        val params = Array[Any](adClickTrend.date, adClickTrend.hour, adClickTrend.minute, adClickTrend.adid)
        client.executeQuery(selectSQL, params, new QueryCallback {
          override def process(rs: ResultSet): Unit = {
            if (rs.next() && rs.getInt(1) > 0) {
              updateAdClickTrends += adClickTrend
            } else {
              insertAdClickTrends += adClickTrend
            }
          }
        })
      }

      // 执行批量更新操作
      // 此处的UPDATE是覆盖
      val updateSQL = "UPDATE ad_click_trend SET clickCount=? " +
        "WHERE date=? " +
        "AND hour=? " +
        "AND minute=? " +
        "AND adid=?"

      val updateParamsList: ArrayBuffer[Array[Any]] = ArrayBuffer[Array[Any]]()

      for (adClickTrend <- updateAdClickTrends) {
        updateParamsList += Array[Any](adClickTrend.clickCount, adClickTrend.date, adClickTrend.hour, adClickTrend.minute, adClickTrend.adid)
      }
      if (updateParamsList.nonEmpty) {
        client.executeBatch(updateSQL, updateParamsList.toArray)
        println(s"AdClickTrendDAO: 成功批量更新 ${updateParamsList.size} 条记录。")
      }


      // 执行批量插入操作
      val insertSQL = "INSERT INTO ad_click_trend VALUES(?,?,?,?,?)"

      val insertParamsList: ArrayBuffer[Array[Any]] = ArrayBuffer[Array[Any]]()

      for (adClickTrend <- insertAdClickTrends) {
        insertParamsList += Array[Any](adClickTrend.date, adClickTrend.hour, adClickTrend.minute, adClickTrend.adid, adClickTrend.clickCount)
      }
      if (insertParamsList.nonEmpty) {
        client.executeBatch(insertSQL, insertParamsList.toArray)
        println(s"AdClickTrendDAO: 成功批量插入 ${insertParamsList.size} 条记录。")
      }

    } catch {
      case e: Exception =>
        println(s"AdClickTrendDAO: 批量更新/插入时发生异常: ${e.getMessage}")
        e.printStackTrace()
    } finally {
      // 确保将 MySqlProxy 实例返回给对象池
      if (client != null) {
        CreateMySqlPool().returnObject(client)
      }
    }
  }

}